package com.study.one.pulsar.producer;

import com.study.one.propertise.OneSettingsProperties;
import com.study.one.pulsar.PulsarClientFactory;
import com.study.one.pulsar.messagedto.OrderMessageDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.schema.JSONSchema;
import org.apache.pulsar.shade.io.netty.util.concurrent.DefaultThreadFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author wu
 */
@Slf4j
@Component
public class PulsarOneProducerInit {


    private final PulsarClientFactory pulsarClientFactory;

    private final OneSettingsProperties oneSettingsProperties;

    public PulsarOneProducerInit(PulsarClientFactory pulsarClientFactory, OneSettingsProperties oneSettingsProperties) {
        this.pulsarClientFactory = pulsarClientFactory;
        this.oneSettingsProperties = oneSettingsProperties;
    }

    private Producer<OrderMessageDTO> producer;

    /**
     * 发送消息到 pulsar
     *
     * @param messageDTO 消息字符串
     */
    public MessageId sendMessage(OrderMessageDTO messageDTO) {
        try {
            log.info("DisputeBackProducer send message to pulsar, message : {}", messageDTO);
            if (messageDTO != null) {
                return producer.send(messageDTO);
            }
        } catch (Exception e) {
            log.error("DisputeBackProducer.send message is error", e);
        }
        return null;
    }

    @PostConstruct
    void init() throws PulsarClientException {
        if (pulsarClientFactory != null) {
            log.info("========PulsarOneProducerInit orderTopic====");
            String orderTopic = oneSettingsProperties.pulsar.getProducerTopic().getOrderTopic();
            log.info("orderTopic:{}",orderTopic);
            producer = pulsarClientFactory.createRouteProducer(orderTopic, JSONSchema.of(OrderMessageDTO.class));
        }
    }

    @PreDestroy
    void destroy() throws PulsarClientException {
        if (null != producer) {
            producer.close();
        }
    }
}
